package ru.unlocker.reports.sampler;

import io.quarkus.runtime.QuarkusApplication;
import io.quarkus.runtime.annotations.QuarkusMain;
import io.smallrye.mutiny.tuples.Tuple2;
import jakarta.inject.Inject;
import java.io.File;
import java.time.Duration;
import java.time.Instant;
import java.time.LocalDate;
import java.util.function.Consumer;
import lombok.SneakyThrows;
import lombok.val;
import org.eclipse.microprofile.reactive.messaging.Channel;
import org.eclipse.microprofile.reactive.messaging.Emitter;
import picocli.CommandLine;
import picocli.CommandLine.Option;
import picocli.CommandLine.Parameters;
import ru.unlocker.reports.domain.ReportAndTimerEvent;
import ru.unlocker.reports.sampler.domain.TopicEnum;
import ru.unlocker.reports.sampler.service.EventDataReader;

@QuarkusMain
@CommandLine.Command(name = "sampler", mixinStandardHelpOptions = true)
public class SamplerApp implements Runnable, QuarkusApplication {

  @Inject
  CommandLine.IFactory factory;

  @Inject
  @Channel("reports")
  Emitter<ReportAndTimerEvent> reportsEmitter;

  @Inject
  @Channel("timers")
  Emitter<ReportAndTimerEvent> timersEmitter;

  @Option(names = {"-t", "--tick"}, description = "Duration time for a one tick", defaultValue = "PT5s")
  Duration tick;

  @Option(names = {"-w", "--workday"}, description = "Date for the workday", defaultValue = "2024-04-22")
  LocalDate workday;

  @Option(names = {"-b", "--bootstrap"}, description = "Bootstrap servers for Kafka", defaultValue = "kafka1:9092")
  String bootstrap;

  @Option(names = {"-r", "--reports-topic"}, description = "Reports topic name", defaultValue = "reports")
  String reportsTopic;

  @Option(names = {"-m", "--timers-topic"}, description = "Timers topic name", defaultValue = "timers")
  String timersTopic;

  @Parameters(description = "CSV file names")
  File[] files;

  @Override
  public int run(String... args) {
    return new CommandLine(this, factory).execute(args);
  }

  @Override
  public void run() {
    val zeroTime = Instant.now();
    for (File file : files) {
      try {
        val sourceEvents = EventDataReader.readCsv(file);
        sourceEvents.stream()
            .map(s ->
                Tuple2.of(
                    s.getTopic(),
                    ReportAndTimerEvent.builder()
                        .createdAt(zeroTime.plus(tick.multipliedBy(s.getTicks())))
                        .reportCode(s.getReportCode())
                        .workday(workday)
                        .reportId(s.getReportId())
                        .groupId(s.getGroupId())
                        .errorMessage(s.getErrorDesc())
                        .duration(s.getDuration())
                        .reportEvent(s.getReportKind())
                        .timerEvent(s.getTimerKind())
                        .build()
                )
            )
            .forEach(this::accept);
      } catch (Exception e) {
        System.err.println("Error while processing file: " + file.getAbsolutePath() + ": " + e.getMessage());
      }
    }
  }

  @SneakyThrows
  private void accept(Tuple2<TopicEnum, ReportAndTimerEvent> t) {
    switch (t.getItem1()) {
      case T -> timersEmitter.send(t.getItem2());
      case R -> reportsEmitter.send(t.getItem2());
    }
  }
}
